import beans.OrderEvent;
import beans.OrderResult;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;

/**
 * @author zkq
 * @date 2022/10/15 16:35
 */
public class OrderTimeout {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<OrderEvent> orderStream = env.readTextFile("F:\\javasecode220620\\UserBehaviorAnalysis\\OrderTimeoutDetect\\src\\main\\resources\\OrderLog.csv")
                .map(line -> {
                    String[] split = line.split(",");
                    return new OrderEvent(new Long(split[0]), split[1], split[2], new Long(split[3]));
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<OrderEvent>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderEvent>() {
                            @Override
                            public long extractTimestamp(OrderEvent element, long recordTimestamp) {
                                return element.getTimestamp() * 1000;
                            }
                        })
                );

        Pattern<OrderEvent, OrderEvent> orderPayPattern = Pattern.<OrderEvent>begin("create")
                .where(new SimpleCondition<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent value) throws Exception {
                        return "create".equals(value.getEventType());
                    }
                })
                //选择宽松近邻，因为下单和支付之间可以随便修改 不需要严格
                .followedBy("pay")
                .where(new SimpleCondition<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent value) throws Exception {
                        return "pay".equals(value.getEventType());
                    }
                })
                .within(Time.minutes(15));

        //定义侧输出流标签，因为我们实际检测的是匹配上了create但是超时未匹配上pay的数据，即匹配上了模式的一部分，所以需要侧输出流
        OutputTag<OrderResult> orderTimeoutTag = new OutputTag<OrderResult>("order-timeout"){};

        //将pattern应用到输入数据流上，得到patternStream
        PatternStream<OrderEvent> patternStream = CEP.pattern(orderStream.keyBy(OrderEvent::getOrderId), orderPayPattern);

        //调用select方法，实现对匹配复杂时间和超时复杂事件的提取和处理
        SingleOutputStreamOperator<OrderResult> resultStream = patternStream.select(orderTimeoutTag, new OrderTimeoutSelect(), new OrderPaySelect());

        resultStream.print("payed normally");
        resultStream.getSideOutput(orderTimeoutTag)
                .print("timeout");

        env.execute("order timeout detect job");
    }

    public static class OrderTimeoutSelect implements PatternTimeoutFunction<OrderEvent,OrderResult>{

        @Override
        public OrderResult timeout(Map<String, List<OrderEvent>> pattern, long timeoutTimestamp) throws Exception {
            Long timeoutOrderId = pattern.get("create").get(0).getOrderId();
            return new OrderResult(timeoutOrderId,"timeout " + timeoutTimestamp);
        }
    }

    public static class  OrderPaySelect implements PatternSelectFunction<OrderEvent,OrderResult>{

        @Override
        public OrderResult select(Map<String, List<OrderEvent>> pattern) throws Exception {
            Long payedOrderId = pattern.get("pay").get(0).getOrderId();
            return new OrderResult(payedOrderId,"payed" );
        }
    }
}
